/**
 * Copyright (C) 2008 by The Regents of the University of California
 * Redistribution of this file is permitted under the terms of the GNU
 * Public License (GPL).
 *
 * @author Junghoo "John" Cho <cho AT cs.ucla.edu>
 * @date 3/24/2008
 */

#include <cstdio>
#include <iostream>
#include <fstream>
#include "Bruinbase.h"
#include "SqlEngine.h"
#include "BTreeIndex.h"

using namespace std;

// external functions and variables for load file and sql command parsing 
extern FILE* sqlin;
int sqlparse(void);


RC SqlEngine::run(FILE* commandline)
{
  fprintf(stdout, "Bruinbase> ");

  // set the command line input and start parsing user input
  sqlin = commandline;
  sqlparse();  // sqlparse() is defined in SqlParser.tab.c generated from
               // SqlParser.y by bison (bison is GNU equivalent of yacc)

  return 0;
}

RC SqlEngine::select(int attr, const string& table, const vector<SelCond>& cond)
{
  RecordFile rf;   // RecordFile containing the table
  RecordId   rid;  // record cursor for table scanning
  BTreeIndex idx;  // BTreeIndex

  RC     rc;
  int    key;     
  string value;
  int    count;
  int    diff;
  IndexCursor cursor;
  
  rid.pid = rid.sid = 0;
  count = 0;
  
  bool exitLoop = false;

  bool useTable = true;
  bool hasMinKey = false;
  bool hasMaxKey = false;
  bool hasNotEqualKeys = false;
  int minK;  // GE
  int maxK;  // LE
  vector<int> notEqKeys;
  
  bool hasValConstraints = false;
  bool hasMinV = false;
  bool hasMaxV = false;
  bool hasNotEqVals = false;
  char minV[100]; // GE
  char maxV[100]; // LE
  vector<char*> notEqVals;
    
  
  for (unsigned i = 0; i < cond.size(); i++) {
    switch (cond[i].attr) {
      case 1: // Key
        switch (cond[i].comp) {
          case SelCond::EQ:
            if (hasMinKey && hasMaxKey) {
	      if (minK <= atoi(cond[i].value) && maxK >= atoi(cond[i].value)) { 
		minK = maxK = atoi(cond[i].value);
	      }
	      else
		goto exit_select;
            }
	    else if (hasMinKey) {
	      if (minK <= atoi(cond[i].value)) {
		minK = maxK = atoi(cond[i].value);
		hasMaxKey = true;
	      }
	      else
		goto exit_select;
	    }
	    else if (hasMaxKey) {
	      if (maxK >= atoi(cond[i].value)) {
		minK = maxK = atoi(cond[i].value);
		hasMinKey = true;
	      }
	      else
		goto exit_select;
	    }
	    else {
	      minK = maxK = atoi(cond[i].value);
	      hasMinKey = hasMaxKey = true;
	    }
            break; 
          case SelCond::NE:
	    hasNotEqualKeys = true;
            notEqKeys.push_back(atoi(cond[i].value));
            break;
          case SelCond::LT:
            if (!hasMaxKey) {
              maxK = atoi(cond[i].value) - 1;
	      hasMaxKey = true;
            }
            else {
              if (maxK >= atoi(cond[i].value))
                maxK = atoi(cond[i].value) - 1;
            }
            break;
          case SelCond::GT:
            if (!hasMinKey) {
              minK = atoi(cond[i].value) + 1;
	      hasMinKey = true;
            }
            else {
              if (minK <= atoi(cond[i].value))
                minK = atoi(cond[i].value) + 1;
            }
            break;
          case SelCond::LE:
            if (!hasMaxKey) {
              maxK = atoi(cond[i].value);
	      hasMaxKey = true;
            }
            else {
              if (maxK > atoi(cond[i].value))
                maxK = atoi(cond[i].value);
            }
            break;
          case SelCond::GE:
            if (!hasMinKey) {
              minK = atoi(cond[i].value);
	      hasMinKey = true;
            }
            else {
              if (minK < atoi(cond[i].value))
                minK = atoi(cond[i].value);
            }
            break;
        }
        break;
      case 2: // Value
        switch (cond[i].comp) {
          case SelCond::EQ:
            if (hasMinV && hasMaxV) {
	      if(strcmp(minV, cond[i].value) < 1 && strcmp(maxV, cond[i].value) > -1) {
		strcpy(minV, cond[i].value);
		strcpy(maxV, cond[i].value);
	      }
	      else
		goto exit_select;
            }
            else if (hasMinV) {
	      if (strcmp(minV, cond[i].value) < 1) {
		strcpy(minV, cond[i].value);
		strcpy(maxV, cond[i].value);
		hasMaxV = true;
	      }
	      else
		goto exit_select;
	    }
	    else if (hasMaxV) {
	      if (strcmp(maxV, cond[i].value) > -1) {
		strcpy(minV, cond[i].value);
		strcpy(maxV, cond[i].value);
		hasMinV = true;
	      }
	      else
		goto exit_select;
	    }
	    else {
	      strcpy(minV, cond[i].value);
	      strcpy(maxV, cond[i].value);
	      hasMaxV = hasMinV = true;
	      hasValConstraints = true;
	    }
            break; 
           case SelCond::NE:
	     notEqVals.push_back(cond[i].value);
	     hasNotEqVals = true;
	     hasValConstraints = true;
	     break;
          case SelCond::LT:
            if (!hasMaxV) {
              strcpy(maxV, cond[i].value);
	      maxV[strlen(maxV)-1] -= 1;
	      hasMaxV = true;
	      hasValConstraints = true;
            }
            else {
              if (strcmp(maxV, cond[i].value) > -1) {
                strcpy(maxV, cond[i].value);
		maxV[strlen(maxV)-1] -= 1;
	      }
            }
            break;
          case SelCond::GT:
            if (!hasMinV) {
              strcpy(minV, cond[i].value);
	      minV[strlen(minV)-1] += 1;
              hasMinV = true;
	      hasValConstraints = true;
            }
            else {
              if (strcmp(minV, cond[i].value) < 1) {
                strcpy(minV, cond[i].value);
		minV[strlen(minV)-1] += 1;
	      }
            }
            break;
          case SelCond::LE:
            if (!hasMaxV) {
              strcpy(maxV, cond[i].value);
	      hasMaxV = true;
	      hasValConstraints = true;
            }
            else {
              if (strcmp(maxV, cond[i].value) == 1)
                strcpy(maxV, cond[i].value);
            }
            break;
          case SelCond::GE:
            if (!hasMinV) {
              strcpy(minV, cond[i].value);
	      hasMinV = true;
	      hasValConstraints = true;
            }
            else {
              if (strcmp(minV, cond[i].value) == -1)
                strcpy(minV, cond[i].value);
            }
            break;
        }
        break;
    }
  }

  if(hasValConstraints || attr == 2 || attr == 3) {
    // open the table file
    if ((rc = rf.open(table + ".tbl", 'r')) < 0) {
      fprintf(stderr, "Error: table %s does not exist\n", table.c_str());
      return rc;
    }
  }

  if (hasMinKey || hasMaxKey || (attr == 4 && !hasValConstraints)) {
    if (idx.open(table + ".idx", 'r') == 0) {
      useTable = false;
      if (hasMinKey && hasMaxKey) {
	if(idx.locate(minK, cursor) == RC_NO_SUCH_RECORD)
	  goto exit_select;
	while(idx.readForward(cursor, key, rid) == 0 && key <= maxK && key >= minK) {
	  for(int i = 0; i < notEqKeys.size(); i++) {
	    if(notEqKeys[i] == key) {
	      exitLoop = true;
	      break;
	    }
	  }
	  if (exitLoop) {
	    exitLoop = false;
	    continue;
	  }
	  if(hasValConstraints || attr == 2 || attr == 3) {
	    if (rf.read(rid, key, value) < 0) {
	      fprintf(stderr, "Error: while reading a tuple from table %s\n", table.c_str());
	      goto exit_select;
	    }
	  }
	  if(hasValConstraints) {
	    if(hasMinV) {
	      if(strcmp(value.c_str(), minV) == -1) {
		continue;
	      }
	    }
	    if(hasMaxV) {
	      if(strcmp(value.c_str(), maxV) == 1) {
		continue;
	      }
	    }
	    if(hasNotEqVals) {
	      for(int i = 0; i < notEqVals.size(); i++) {
		if(strcmp(notEqVals[i], value.c_str()) == 0) {
		  continue;
		}
	      }
	    }
	  }
	  switch(attr) {
	  case 1:  // SELECT key
	    fprintf(stdout, "%d\n", key);
	    break;
	  case 2:
	    fprintf(stdout, "%s\n", value.c_str());
	    break;
	  case 3:
	    fprintf(stdout, "%d '%s'\n", key, value.c_str());
	    break;
	  case 4:
	    count++;
	    break;
	  }
	}
      }
      else if (hasMinKey) {
	if(idx.locate(minK, cursor) == RC_NO_SUCH_RECORD)
	  goto exit_select;
	while(idx.readForward(cursor, key, rid) != RC_INVALID_PID) {
	  for(int i = 0; i < notEqKeys.size(); i++) {
	    if(notEqKeys[i] == key) {
	      exitLoop = true;
	      break;
	    }
	  }
	  if(exitLoop) {
	    exitLoop = false;
	    continue;
	  }
	  if(hasValConstraints || attr == 2 || attr == 3) {
	    if (rf.read(rid, key, value) < 0) {
	      fprintf(stderr, "Error: while reading a tuple from table %s\n", table.c_str());
	      goto exit_select;
	    }
	  }
	  if(hasValConstraints) {
	    if(hasMinV) {
	      if(strcmp(value.c_str(), minV) == -1) {
		continue;
	      }
	    }
	    if(hasMaxV) {
	      if(strcmp(value.c_str(), maxV) == 1) {
		continue;
	      }
	    }
	    if(hasNotEqVals) {
	      for(int i = 0; i < notEqVals.size(); i++) {
		if(strcmp(notEqVals[i], value.c_str()) == 0) {
		  exitLoop = true;
		  break;
		}
	      }
	      if(exitLoop) {
		exitLoop = false;
		continue;
	      }
	    }
	  }
	  switch(attr) {
	  case 1:  // SELECT key
	    fprintf(stdout, "%d\n", key);
	    break;
	  case 2:
	    fprintf(stdout, "%s\n", value.c_str());
	    break;
	  case 3:
	    fprintf(stdout, "%d '%s'\n", key, value.c_str());
	    break;
	  case 4:
	    count++;
	    break;
	  }
	}
      }
      else if (hasMaxKey) {
	if(idx.locate(0, cursor) == RC_NO_SUCH_RECORD)
	  goto exit_select;
	while(idx.readForward(cursor, key, rid) != RC_INVALID_PID && key < maxK) {
	  for(int i = 0; i < notEqKeys.size(); i++) {
	    if(notEqKeys[i] == key) {
	      exitLoop = true;
	      break;
	    }
	  }
	  if(exitLoop) {
	    exitLoop = false;
	    continue;
	  }
	  if(hasValConstraints || attr == 2 || attr == 3) {
	    if (rf.read(rid, key, value) < 0) {
	      fprintf(stderr, "Error: while reading a tuple from table %s\n", table.c_str());
	      goto exit_select;
	    }
	  }
	  if(hasValConstraints) {
	    if(hasMinV) {
	      if(strcmp(value.c_str(), minV) == -1) {
		continue;
	      }
	    }
	    if(hasMaxV) {
	      if(strcmp(value.c_str(), maxV) == 1) {
		continue;
	      }
	    }
	    if(hasNotEqVals) {
	      for(int i = 0; i < notEqVals.size(); i++) {
		if(strcmp(notEqVals[i], value.c_str()) == 0) {
		  exitLoop = true;
		  break;
		}
	      }
	      if(exitLoop) {
		exitLoop = false;
		continue;
	      }
	    }
	  }
	  switch(attr) {
	  case 1:  // SELECT key
	    fprintf(stdout, "%d\n", key);
	    break;
	  case 2:
	    fprintf(stdout, "%s\n", value.c_str());
	    break;
	  case 3:
	    fprintf(stdout, "%d '%s'\n", key, value.c_str());
	    break;
	  case 4:
	    count++;
	    break;
	  }
	}
      }
      else {
	if(idx.locate(0, cursor) == RC_NO_SUCH_RECORD)
	  goto exit_select;
	while(idx.readForward(cursor, key, rid) != RC_INVALID_PID) {
	  for(int i = 0; i < notEqKeys.size(); i++) {
	    if(notEqKeys[i] == key) {
	      exitLoop = true;
	      break;
	    }
	  }
	  if(exitLoop) {
	    exitLoop = false;
	    continue;
	  }
	  count++;
        }
      }
    }
  }

  if (useTable) {
    if(!hasValConstraints && !(attr == 2) && !(attr == 3)) {
      // open the table file
      if ((rc = rf.open(table + ".tbl", 'r')) < 0) {
	fprintf(stderr, "Error: table %s does not exist\n", table.c_str());
	return rc;
      }
    }
    // scan the table file from the beginning
    while (rid < rf.endRid()) {
      // read the tuple
      if ((rc = rf.read(rid, key, value)) < 0) {
        fprintf(stderr, "Error: while reading a tuple from table %s\n", table.c_str());
        goto exit_select;
      }
    
      if(hasMinKey) {
	if(key < minK)
	  goto next_tuple;
      }
      if(hasMaxKey) {
	if(key > maxK)
	  goto next_tuple;
      }
      for(int i = 0; i < notEqKeys.size(); i++) {
	if(key == notEqKeys[i])
	  goto next_tuple;
      }
      if(hasMinV) {
	if(value < minV)
	  goto next_tuple;
      }
      if(hasMaxV) {
	if(value > maxV)
	  goto next_tuple;
      }
      for(int i = 0; i < notEqVals.size(); i++) {
	if(value == notEqVals[i])
	  goto next_tuple;
      }
      // the condition is met for the tuple. 
      // increase matching tuple counter
      count++;

      // print the tuple 
      switch (attr) {
        case 1:  // SELECT key
          fprintf(stdout, "%d\n", key);
          break;
        case 2:  // SELECT value
          fprintf(stdout, "%s\n", value.c_str());
          break;
        case 3:  // SELECT *
          fprintf(stdout, "%d '%s'\n", key, value.c_str());
          break;
      }

      // move to the next tuple
      next_tuple:
      ++rid;
    }
  }

  // print matching tuple count if "select count(*)"
  if (attr == 4) {
    fprintf(stdout, "%d\n", count);
  }
  rc = 0;
  
  // close the table file and return
  exit_select:
  
  rf.close();
  return rc;
}

RC SqlEngine::load(const string& table, const string& loadfile, bool index)
{
	RecordFile tableRF = RecordFile(table + ".tbl", 'w');
	FILE* loadInput = fopen(loadfile.c_str(), "r"); 
	char fileLine[128];
    
    // Check if indexing is needed
	BTreeIndex* BIndex = NULL;
    if (index) {
        BIndex = new BTreeIndex();
        BIndex->open(table + ".idx", 'w');
    }
    
	if (loadInput == NULL) {
		return RC_FILE_OPEN_FAILED;
	}
	else {
		while (fgets(fileLine, 128, loadInput) != NULL) {
			int tupleKey;
			string tupleValue;
			RecordId rid;
			parseLoadLine(fileLine, tupleKey, tupleValue);
			tableRF.append(tupleKey, tupleValue, rid);
            if (index) 
                BIndex->insert(tupleKey, rid);
		}
		fclose(loadInput);
	}
	tableRF.close();
    if (index) {
        BIndex->writeIndexData();
        BIndex->close();
        delete BIndex;
    }
    return 0;
}

RC SqlEngine::parseLoadLine(const string& line, int& key, string& value)
{
    const char *s;
    char        c;
    string::size_type loc;
    
    // ignore beginning white spaces
    c = *(s = line.c_str());
    while (c == ' ' || c == '\t') { c = *++s; }

    // get the integer key value
    key = atoi(s);

    // look for comma
    s = strchr(s, ',');
    if (s == NULL) { return RC_INVALID_FILE_FORMAT; }

    // ignore white spaces
    do { c = *++s; } while (c == ' ' || c == '\t');
    
    // if there is nothing left, set the value to empty string
    if (c == 0) { 
        value.erase();
        return 0;
    }

    // is the value field delimited by ' or "?
    if (c == '\'' || c == '"') {
        s++;
    } else {
        c = '\n';
    }

    // get the value string
    value.assign(s);
    loc = value.find(c, 0);
    if (loc != string::npos) { value.erase(loc); }

    return 0;
}
